package com.zyh.study.service;


import com.fasterxml.jackson.databind.ObjectMapper;
import com.zyh.study.config.AlpacaSseProperties;
import com.zyh.study.pojo.JournalEvent;
import com.zyh.study.pojo.TradeEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;
import reactor.util.retry.RetryBackoffSpec;

import java.time.Duration;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @Classname AlpacaSseService
 * @Description TODO()
 * @Date 2025-09-05 2:17
 * @Created by 86177
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class AlpacaSseService {
    private final WebClient webClient;
    private final AlpacaSseProperties properties;
    private final ObjectMapper objectMapper;
    private final ReactiveRedisTemplate<String, String> redisTemplate;

    // 分别记录两种事件类型的最后ID
    private final AtomicReference<String> lastJournalId = new AtomicReference<>();
    private final AtomicReference<String> lastTradeId = new AtomicReference<>();

    /**
     * 获取Journal事件流 - 共享连接
     */
    public Flux<JournalEvent> streamJournalEvents() {
        return createEventStream(properties.getJournalUrl(), "journal", JournalEvent.class, lastJournalId)
                .doOnSubscribe(sub -> log.info("Journal stream subscribed"))
                .doOnCancel(() -> log.info("Journal stream unsubscribed"))
                .share(); // 使用share()实现连接共享
    }

    /**
     * 获取Trade事件流 - 共享连接
     */
    public Flux<TradeEvent> streamTradeEvents() {
        return createEventStream(properties.getTradeUrl(), "trade", TradeEvent.class, lastTradeId)
                .doOnSubscribe(sub -> log.info("Trade stream subscribed"))
                .doOnCancel(() -> log.info("Trade stream unsubscribed"))
                .share(); // 使用share()实现连接共享
    }

    /**
     * 通用事件流创建方法
     */
    private <T> Flux<T> createEventStream(String url, String streamType,
                                          Class<T> eventClass, AtomicReference<String> lastIdRef) {
        return Flux.defer(() -> {
            // 构建查询参数
            String lastId = lastIdRef.get();
            MultiValueMap<String, String> params = new LinkedMultiValueMap<>();

            if (lastId != null) {
                params.add("since_id", lastId);
            } else {
                // 首次连接获取最近5分钟的事件
                String startTime = Instant.now().minus(5, ChronoUnit.MINUTES).toString();
                params.add("since", startTime);
            }

            return webClient.get()
                    .uri(uriBuilder -> uriBuilder
                            .path(url)
                            .queryParams(params)
                            .build())
                    .retrieve()
                    .bodyToFlux(String.class)
                    .name(streamType + "-sse")
                    .onBackpressureBuffer(1000)
                    .flatMap(eventStr -> parseEvent(eventStr, eventClass, streamType))
                    .doOnNext(event -> updateLastId(event, lastIdRef))
                    .doOnSubscribe(sub -> log.info("Starting {} SSE connection", streamType))
                    .doOnCancel(() -> log.warn("{} SSE connection cancelled", streamType))
                    .doOnTerminate(() -> log.error("{} SSE connection terminated", streamType))
                    .doOnError(e -> log.error("{} SSE connection error", streamType, e))
                    .retryWhen(createRetryStrategy());
        });
    }

    /**
     * 事件解析（支持注释行过滤）
     */
    private <T> Mono<T> parseEvent(String eventStr, Class<T> eventClass, String streamType) {
        return Mono.fromCallable(() -> {
            // 跳过注释行
            if (eventStr.startsWith(":")) {
                handleCommentMessage(eventStr.substring(1).trim(), streamType);
                return null;
            }

            T event = objectMapper.readValue(eventStr, eventClass);

            // 验证必要字段
            if (event instanceof JournalEvent) {
                JournalEvent je = (JournalEvent) event;
                if (je.getEventId() == null || je.getJournalId() == null) {
                    log.warn("Invalid journal event: {}", eventStr);
                    return null;
                }
            } else if (event instanceof TradeEvent) {
                TradeEvent te = (TradeEvent) event;
                if (te.getEventId() == null || te.getOrder() == null) {
                    log.warn("Invalid trade event: {}", eventStr);
                    return null;
                }
            }

            return event;
        }).onErrorResume(e -> {
            log.error("Failed to parse {} event: {}", streamType, eventStr, e);
            return Mono.empty();
        });
    }

    /**
     * 处理Alpaca的特殊注释消息
     */
    private void handleCommentMessage(String message, String streamType) {
        if (message.startsWith("slow client")) {
            log.warn("Alpaca {} stream warning: {}", streamType, message);
        } else if (message.startsWith("internal server error")) {
            log.error("Alpaca {} stream server error: {}", streamType, message);
        } else {
            log.debug("Alpaca {} stream comment: {}", streamType, message);
        }
    }

    /**
     * 更新最后事件ID
     */
    private <T> void updateLastId(T event, AtomicReference<String> lastIdRef) {
        String id = null;

        if (event instanceof JournalEvent) {
            id = ((JournalEvent) event).getEventId();
        } else if (event instanceof TradeEvent) {
            id = ((TradeEvent) event).getEventId();
        }

        if (id != null) {
            lastIdRef.set(id);
            log.debug("Updated last event ID: {}", id);

            // 可选：保存到Redis
            redisTemplate.opsForValue().set("last_id:" + lastIdRef, id, Duration.ofHours(1))
                    .subscribe(r -> log.trace("Saved last ID to Redis"));
        }
    }

    /**
     * 重试策略（保持原有实现）
     */
    private RetryBackoffSpec createRetryStrategy() {
        return Retry.backoff(properties.getReconnect().getMaxAttempts(),
                        Duration.ofMillis(properties.getReconnect().getInitialDelay()))
                .maxBackoff(Duration.ofMillis(properties.getReconnect().getMaxDelay()))
                .jitter(0.5)
                .doBeforeRetry(retry -> log.warn("Retry attempt #{}/{} after failure: {}",
                        retry.totalRetries() + 1,
                        properties.getReconnect().getMaxAttempts(),
                        retry.failure().getMessage()))
                .onRetryExhaustedThrow((spec, signal) ->
                        new RuntimeException("Max retry attempts exceeded"));
    }
}
